package com.study.flink.datastream

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.core.fs.FileSystem.WriteMode
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ArrayBuffer

/**
  * nc -lk 9999
  * 时间戳 单词
  * EventTime Demo （存在问题）
  */
object FlinkEventTimeDemo {

  def main(args: Array[String]): Unit = {
    // 1、初始化环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 2、读取数据
    val data = env.socketTextStream("192.168.11.11", 9999)

    // 3、处理数据
    import org.apache.flink.api.scala._
    data.map(x => {
      val splits = x.split("\\s")
      println(x)
      (splits(0).toLong, splits(1))
    }).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[(Long, String)] {
      // 最大无序容忍时间
      val maxOutOfOrderness = 60000L // 60 seconds
      // 当前的最大时间戳
      var currentMaxTimestamp: Long = _

      override def getCurrentWatermark: Watermark = {
        // return the watermark as current highest timestamp minus the out-of-orderness bound
        //println(s"currentMaxTimestamp => $currentMaxTimestamp, maxOutOfOrderness => $maxOutOfOrderness")
        new Watermark(currentMaxTimestamp - maxOutOfOrderness)
      }

      override def extractTimestamp(element: (Long, String), previousElementTimestamp: Long): Long = {
        val timestamp = element._1
        //println(s"timestamp => $timestamp, currentMaxTimestamp => $currentMaxTimestamp")
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp)
        //println(s"new currentMaxTimestamp => $currentMaxTimestamp")
        timestamp
      }
    }).keyBy(1)
      .window(TumblingEventTimeWindows.of(Time.seconds(60)))
      // 参数：输入参数类型，输出参数类型
      .apply(new WindowFunction[(Long, String), (String, String, Long), Tuple, TimeWindow] {
      override def apply(key: Tuple, window: TimeWindow, input: Iterable[(Long, String)], out: Collector[(String, String, Long)]): Unit = {
        val word = key.getField(1).toString
        println(word)
        // 求单词总次数
        var sum = 0L
        val times = ArrayBuffer[Long]()
        val iterator = input.iterator
        while (iterator.hasNext) {
          val next = iterator.next()
          sum += 1
          times.append(next._1)
        }

        val time = new SimpleDateFormat("yyyy-MM-dd HH:mm").format(new Date(times.max))
        // 日期，单词，次数
        out.collect((time, word, 0L))
      }
    }).writeAsText("d://result.TXT", WriteMode.OVERWRITE).setParallelism(1)

    // 4、触发任务执行
    env.execute("Flink EventTime Demo")

  }
}
